/* Copyright  2009-2010
   @Author
   Richard Changde Yin            e-mail yinchangde@hotmail.com

   This program is free software; you can redistribute it and/or modify
   it under the terms of the GNU General Public License as published by
   the Free Software Foundation; version 2 of the License.

   This program is distributed in the hope that it will be useful,
   but WITHOUT ANY WARRANTY; without even the implied warranty of
   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
   GNU General Public License for more details.

   You should have received a copy of the GNU General Public License
   along with this program; if not, write to the Free Software
   Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA */

/**
 * @filename: catalogTables.cc
 *
 */

#include <server_includes.h>


#include <table.h>
#include <session.h>
#include <field/timestamp.h>
#include <field/varstring.h>

#include<catalogtables.h>
#include<database.h>
#include<allocator.h>
#include<mem_field.h>

#include <gettext.h>
#include <error.h>
#include <errmsg_print.h>
#include<debug.h>



using namespace BackEnd ;

char ChunkName[MAX_CHUNKS][CHUNK_NAME_LEN]={"UserChunkTableId",
                                            "LockTableHashBucketId",
                                            "LockTableMutexId",
                                            "LockTableId",
                                            "TransHasTableId",
                                            "UndoLogTableId",
                                            "",
                                            "",
                                            "",
                                            "",
                                            "DatabaseTableId",
                                            "UserTableId",
                                            "TableTableId",
                                            "FieldTableId",
                                            "AccessTableId",
                                            "IndexTableId",
                                            "IndexFieldTableId",
                                            "UserIndexChunkId",
                                            ""};

#define CLEAN_CFIELD_PTR(DB)           do {\
          BackEnd::ListIterator iter = tblfieldptr->fldList.getIterator();\
          void* _data =0;\
            while(iter.hasElement())\
           {\
             _data = iter.nextElement();\
              errmsg_printf(ERRMSG_LVL_DBUG,\
              _("[%s]: line: [%d] One Row deleted from CFIELD table: [%p]\n"),__FILE__,__LINE__,_data);\
           }\
          }while(0)
BackEnd::DbRetVal CatalogTableTABLE::insert(const char *tbl_name,
                                              int64_t ,
                                              size_t  size,
                                              int32_t numFlds,
                                              void*  chunk,
                                              void * &tptr,
                                              int32_t namelen ,
                                              HA_CREATE_INFO* create_info,
                                              TABLETYPE _tbltype)
{

#define  CTABLE_CLEANUP  do { \
                           delete tblfieldptr ;\
                           tChunk->free(systemDatabase_, tptr);\
                           tptr = 0;\
                           free((char*) tblnameptr); tblnameptr=0;\
                           return BackEnd::ER_SysInternal ;\
                         }while(0)


    struct tblname_ptr_struct* tblnameptr = (struct tblname_ptr_struct*)
                                            malloc(sizeof( struct tblname_ptr_struct));
    if(!tblnameptr) return (BackEnd::DbRetVal)HA_ERR_OUT_OF_MEM ;
    bzero((char*)tblnameptr,sizeof(struct tblname_ptr_struct));
    Chunk *tChunk = systemDatabase_->getSystemDatabaseChunk(TableTableId);
    BackEnd::DbRetVal rv = BackEnd::OK;
    tptr = tChunk->allocate(systemDatabase_, &rv);
      if(!tptr)
    {
      errmsg_printf(ERRMSG_LVL_DBUG,
                   _("Fail to allocate memory for CTABLE chunk: [%s].\n"),tbl_name);
      return rv;
    }
    // CTABLE
    CTABLE *tableInfo = (CTABLE*)tptr;
    tableInfo->init();
    tableInfo->initLock();
    memcpy(tableInfo->tblName_,  tbl_name,namelen);
    memcpy(tblnameptr->tblName_,tbl_name,namelen);
    /* fill in the tblname_ptr_struct structure */
    tblnameptr->keylength   = namelen ;
    tblnameptr->chunkptr    = chunk   ;
    tblnameptr->tblptr      = tptr    ;

    Tbl_Field* tblfieldptr = new Tbl_Field();
    sprintf(tblfieldptr->add_str,"%p",tptr);


    /* Start to insert into name hash*/
    if ( systemDatabase_->insertTblNamePtr2Hash((unsigned char*)tblnameptr))
    {
      //Some error , We should delete tptr.
      errmsg_printf(ERRMSG_LVL_DBUG,
                    _("[%s] line: [%d] Fail to insert [%p] table name:[%s] into Hashtable.\n"),
                                               __FILE__,__LINE__,
                                               tblnameptr,tbl_name);
      CTABLE_CLEANUP;
    }


    if(systemDatabase_->insertTblFieldPtr2Hash((unsigned char*)tblfieldptr))
    {
      CLEAN_CFIELD_PTR(systemDatabase_) ;
      errmsg_printf(ERRMSG_LVL_WARN,
                 _("Fail to insert CFIELD entry into hashtable: [%p]\n"),tblfieldptr);

      CTABLE_CLEANUP;
    }

    tableInfo->tblID_          = (void*)tblnameptr   ;
    tableInfo->length_of_tuple = size ;
    tableInfo->numFlds_        = numFlds;
    tableInfo->numIndexes_     = 0    ;
    tableInfo->chunkPtr_       = chunk;
    tableInfo->table_charset   = create_info->table_charset ;
    tableInfo->namelength      = namelen;
    tableInfo->tbl_type        = _tbltype ;
    tableInfo->auto_increment = (create_info->auto_increment_value ?
                                 create_info->auto_increment_value - 1 : 0);
    errmsg_printf(ERRMSG_LVL_DBUG,
                _("[%s]: line: [%d] One Row inserted into CTABLE catalog "
                                                                      "table: address [%p] "
                                                                      "table name [%s].\n"),
                                                                      __FILE__,__LINE__,
                                                                       tblnameptr,tbl_name);
    return  BackEnd::OK;
}

BackEnd::DbRetVal CatalogTableTABLE::remove(const char *name,
                                             void *&chunk,
                                             void *&tptr)
{

    struct tblname_ptr_struct* tblnameptr = (struct tblname_ptr_struct*)
                  systemDatabase_->searchTblNamePtr((unsigned char*)name,strlen(name));

    if(!tblnameptr)
    {
      errmsg_printf(ERRMSG_LVL_WARN,
       _("Fail to find entry from CTABLE hash table:table name [%s]\n"),name);
      return BackEnd::ER_NotExists;
    }
    void *data = 0;
    Chunk *tChunk = systemDatabase_->getSystemDatabaseChunk(TableTableId);
    data=tptr= tblnameptr->tblptr;
    chunk  = tblnameptr->chunkptr;

    if(systemDatabase_->removeTblNamePtrFromHash((unsigned char*)name,strlen(name)))
    {
      errmsg_printf(ERRMSG_LVL_WARN,
                   _("Fail to delete from CTABLE Hashtable: [%p] table name [%s]\n"), tptr,name);
    }
    if (data)
    {
      CTABLE *ctable =(CTABLE*)tptr;
      ctable->destoryLock();
      tChunk->free(systemDatabase_, tptr);
      errmsg_printf(ERRMSG_LVL_DBUG,
        _("[%s]: line: [%d] One row deleted from CTABLE catalog table:Address [%p] "
                                                           "Table name [%s].\n"),
                                                           __FILE__,__LINE__,
                                                           data,name);
      char _address[BUFFERSIZE];
      sprintf(_address,"%p",tptr);
        if(systemDatabase_->removeTblFieldPtrFromHash((unsigned char*)_address,
                                                     (uint32_t)strlen(_address)))
      {
         errmsg_printf(ERRMSG_LVL_WARN,
                  _("Fail to delete from CFIELD hashtable: [%p]\n"), tptr);
      }else
      {
         errmsg_printf(ERRMSG_LVL_DBUG,
                      _("[%s]: line: [%d] One row deleted from CFIELD hashtable "
                                                              " [%s] \n"),
                                                              __FILE__,__LINE__,
                                                              _address );
      }
    }else
    {
      return BackEnd::ER_NotExists;
    }
    return BackEnd::OK;
}



DbRetVal CatalogTableTABLE::getChunkAndTblPtr(const char *name,
                                               void *&chunk,
                                               void *&tptr)
{
    Chunk *chk = systemDatabase_->getSystemDatabaseChunk(TableTableId);
    /*
     * Search System Database's Hash Table
     */

    struct tblname_ptr_struct* tblnameptr =
                   systemDatabase_->searchTblNamePtr((unsigned char*)name,strlen(name));
    if (tblnameptr){
       chunk = tblnameptr->chunkptr ;
       tptr  = tblnameptr->tblptr   ;
       return BackEnd::OK;
    }
    /// Todo: do we still need this ???
    ChunkIterator iter = chk->getIterator();;
      while (NULL != (tptr = iter.nextElement()))
    {
         if(!my_strcasecmp(table_alias_charset,((CTABLE*)tptr)->tblName_,name))
       {
         //there will be only one row for this table(Primary key)
         chunk = (Chunk*) ((CTABLE*)tptr)->chunkPtr_;
         return BackEnd::OK;
       }
    }
    //CTable was not found
    return BackEnd::ER_NotFound;
}

BackEnd::DbRetVal CatalogTableFIELD::insert(Table &tbl, int , void *tptr)
{
    /*fChunk is pointing to the CFIELD table.*/
    Chunk *fChunk = systemDatabase_->getSystemDatabaseChunk(FieldTableId);

    BackEnd::DbRetVal rv = BackEnd::OK;
    TABLE_SHARE *share = tbl.s ;
    Tbl_Field* tblfieldptr = new Tbl_Field();
    CTABLE* ctable =(CTABLE*) tptr ;
    tblfieldptr->tblptr    = tptr ;
      for (uint32_t fieldnr=0 ; fieldnr < share->fields ; ++fieldnr)
    {
        void *fptr = fChunk->allocate (systemDatabase_, &rv);

        if (!fptr)
        {
          CLEAN_CFIELD_PTR(systemDatabase_) ;
          delete tblfieldptr ;
          errmsg_printf(ERRMSG_LVL_DBUG,
                     _("Fail to allocate for CFIELD catalog table.\n"));
          return rv;
        }
        Field* __field = tbl.field[fieldnr] ;

        unsigned int _type_length = share->fieldnames.type_lengths[fieldnr] ;
        /*CFIELD*/
        CFIELD *fldInfo = (CFIELD*)fptr;

        /* append the fptr here. */
        tblfieldptr->fldList.append(fptr);

        uint32_t cp_length_ = _type_length < IDENTIFIER_LENGTH ?
                                  _type_length : IDENTIFIER_LENGTH ;

        ::memcpy(fldInfo->fldName_,share->fieldnames.type_names[fieldnr],
                                     cp_length_);
        fldInfo->fldNameRealLen = cp_length_ ;

        fldInfo->tblID_  = ctable->tblID_;
        /// table pointer
        fldInfo->tblPtr_ = tptr;

        fldInfo->type_   = __field->type();
        fldInfo->length_ = __field->pack_length()      ;
        fldInfo->offset_ = __field->offset(tbl.record[0]);

        __field->flags & NOT_NULL_FLAG ?
                            fldInfo->isNull_ =false: fldInfo->isNull_  =true ;
        __field->flags & PRI_KEY_FLAG ?
                            fldInfo->isPrimary_=true:fldInfo->isPrimary_=false;
        __field->flags & UNIQUE_KEY_FLAG ?
                            fldInfo->isUnique_= true:fldInfo->isUnique_ =false;
        __field->flags & NO_DEFAULT_VALUE_FLAG ?
                            fldInfo->isDefault_= false:fldInfo->isDefault_=true;
        if ( __field->null_bit)
        {
           fldInfo->null_bit= __field->null_bit;
           fldInfo->null_pos= (uint) (__field->null_ptr -
                                     (unsigned char*) tbl.record[0]);
        }else
        {
           fldInfo->null_bit=  0;
           fldInfo->null_pos=  0;
        }
        if (__field->type() == COLLAPSAR_TYPE_VARCHAR)
        {
            fldInfo->length_bytes=
                         (uint8_t)(((Field_varstring*)__field)->length_bytes);
        }else
        {
            fldInfo->length_bytes= 0;
        }
        fldInfo->charset  = __field->charset() ;
        /*Not used currently*/
        fldInfo->width_ = 0; //TODO
        fldInfo->scale_ = 0; //TODO
        errmsg_printf(ERRMSG_LVL_DBUG,
                 _("[%s]: line: [%d] One Row inserted into CFIELD: field name :[%s]\n"),
                    __FILE__,__LINE__,fldInfo->fldName_);
    } // End of for

     if(systemDatabase_->insertTblFieldPtr2Hash((unsigned char*)tblfieldptr))
    {
      CLEAN_CFIELD_PTR(systemDatabase_) ;
      errmsg_printf(ERRMSG_LVL_WARN,
                    _("Fail to insert CFIELD entry into hashtable: [%p]\n"),tblfieldptr);
      delete tblfieldptr ;
      return BackEnd::ER_SysInternal;
    }
    return BackEnd::OK;
}


BackEnd::DbRetVal CatalogTableFIELD::insert
                                (Table &tbl,
                                 int ,
                                 void *tptr,
                                 char **& out_fptr )// the out parameter
{
      /*fChunk is pointing to the fieldtable.*/
    Chunk *fChunk = systemDatabase_->getSystemDatabaseChunk(FieldTableId);

    BackEnd::DbRetVal rv = BackEnd::OK;
    uint32_t i = 0 ;
    TABLE_SHARE *share = tbl.s ;
    Tbl_Field* tblfieldptr = new Tbl_Field();
    tblfieldptr->tblptr    = tptr ;


    CTABLE* ctable =(CTABLE*) tptr ;
     for (uint32_t fieldnr=0 ; fieldnr < share->fields ; ++fieldnr)
    {
        void *fptr = fChunk->allocate(systemDatabase_, &rv);
         if (!fptr)
        {
          CLEAN_CFIELD_PTR(systemDatabase_) ;
          delete tblfieldptr ;
          errmsg_printf(ERRMSG_LVL_DBUG,
                               _("Fail to allocate for CFIELD catalog table.\n"));
          return rv;
        }
        ///  assign the out parameter.
        out_fptr[i++] = (char*) fptr;
        Field* __field = tbl.field[fieldnr] ;
        unsigned int _type_length = share->fieldnames.type_lengths[fieldnr] ;
        /*CFIELD*/
        CFIELD *fldInfo = (CFIELD*)fptr;
        /* append the fptr */
        tblfieldptr->fldList.append(fptr);
        unsigned int cp_length_ = _type_length < IDENTIFIER_LENGTH ?
                                  _type_length : IDENTIFIER_LENGTH ;

        ::memcpy(fldInfo->fldName_,share->fieldnames.type_names[fieldnr],
                                  cp_length_);
        fldInfo->fldNameRealLen = cp_length_ ;
        fldInfo->charset        = __field->charset();
        fldInfo->tblID_  = ctable->tblID_;
        /// table pointer
        fldInfo->tblPtr_ = tptr;

        fldInfo->type_   = share->field[fieldnr]->type();
        fldInfo->length_ = share->field[fieldnr]->pack_length();
        fldInfo->offset_ = __field->offset(tbl.record[0]);

        share->field[fieldnr]->flags & NOT_NULL_FLAG ?
                            fldInfo->isNull_=false : fldInfo->isNull_   =true ;
        share->field[fieldnr]->flags & PRI_KEY_FLAG ?
                            fldInfo->isPrimary_=true:fldInfo->isPrimary_=false;
        share->field[fieldnr]->flags & UNIQUE_KEY_FLAG ?
                            fldInfo->isUnique_= true:fldInfo->isUnique_ =false;
        share->field[fieldnr]->flags & NO_DEFAULT_VALUE_FLAG ?
                            fldInfo->isDefault_= false:fldInfo->isDefault_=true;
        if ( __field->null_bit)
        {
           fldInfo->null_bit= __field->null_bit;
           fldInfo->null_pos= (uint) (__field->null_ptr -
                                             (unsigned char*) tbl.record[0]);
        }else
        {
           fldInfo->null_bit=  0;
           fldInfo->null_pos=  0;
        }
         if (__field->type() == COLLAPSAR_TYPE_VARCHAR)
         {
           fldInfo->length_bytes=
                                 (uint8_t)(((Field_varstring*)__field)->length_bytes);
         }else
         {
           fldInfo->length_bytes= 0;
         }
        fldInfo->width_ = 0; //TODO
        fldInfo->scale_ = 0; //TODO

        errmsg_printf(ERRMSG_LVL_DBUG,
         _("[%s]: line: [%d] One Row inserted into CFIELD table: field name :[%s]\n"),
                                                                __FILE__, __LINE__,
                                                                fldInfo->fldName_);

    }// End of for

     if(systemDatabase_->insertTblFieldPtr2Hash((unsigned char*)tblfieldptr))
    {
       CLEAN_CFIELD_PTR(systemDatabase_) ;
       errmsg_printf(ERRMSG_LVL_WARN,
                    _("Fail to insert CFIELD entry [%p] into hashtable.\n"),tblfieldptr);
       delete tblfieldptr ;
       return BackEnd::ER_SysInternal;
     }
    return BackEnd::OK;

}

BackEnd::DbRetVal CatalogTableFIELD::remove(void *tptr)
{
    Chunk *fChunk = systemDatabase_->getSystemDatabaseChunk(FieldTableId);
    ChunkIterator fIter = fChunk->getIterator();
    void *data = 0;

    Tbl_Field* tblfieldptr = systemDatabase_->searchTblFieldPtr
                            ((unsigned char*)tptr,(uint32_t)sizeof(tptr));
    if(tblfieldptr) {
      // We go through the list
       BackEnd::ListIterator iter = tblfieldptr->fldList.getIterator();
        while(iter.hasElement())
      {
        data = iter.nextElement();
        fChunk->free(systemDatabase_, data);
        errmsg_printf(ERRMSG_LVL_DBUG,
           _("[%s] line: [%d] One Row deleted from CFIELD table: [%p]\n"),
                                                 __FILE__,__LINE__,data);
      }
    }
    if(systemDatabase_->removeTblFieldPtrFromHash((unsigned char*)tptr,
                                                  (uint32_t)sizeof(tptr)))
    {
      errmsg_printf(ERRMSG_LVL_WARN,
               _("Fail to delete from CFIELD hashtable: [%p]\n"), tptr);
    }

    /*
    while ((data = fIter.nextElement())!= NULL)
    {
        if (((CFIELD*)data)->tblPtr_ == tptr)
        {
            //remove this element
            fChunk->free(systemDatabase_, data);
            printDebug(DM_SystemDatabase,"One Row deleted from FIELD %x",data);
        }
    }
    */
    return BackEnd::OK;
}
/*
void CatalogTableFIELD::getFieldInfo(void* tptr, M_FieldList &list)
{
    Chunk *fChunk = systemDatabase_->getSystemDatabaseChunk(FieldTableId);
    ChunkIterator fIter = fChunk->getIterator();;

    Tbl_Field* tblfieldptr = systemDatabase_->searchTblFieldPtr
                                ((unsigned char*)tptr,(uint32_t)sizeof(tptr));
    if (!tblfieldptr) return ;
    CFIELD* data = 0 ;
    BackEnd::ListIterator iter = tblfieldptr->fldList.getIterator();

    while(iter.hasElement()){
      data =(CFIELD*)iter.nextElement();
      CFIELD *fTuple = (CFIELD*)data;
      m_field_def* fldDef = new m_field_def;

      ::memcpy(fldDef->fldName_,fTuple->fldName_,fTuple->fldNameRealLen) ;
      //fldDef.fldName_[IDENTIFIER_LENGTH] = '\0';
      fldDef->fldNameRealLen    = fTuple->fldNameRealLen ;
      fldDef->type_      = fTuple->type_;
      fldDef->length_    = fTuple->length_;
      fldDef->isDefault_ = fTuple->isDefault_;

           os::memcpy(fldDef.defaultValueBuf_, fTuple->defaultValueBuf_,
                                               DEFAULT_VALUE_BUF_LENGTH);

      fldDef->isNull_    = fTuple->isNull_;
      fldDef->isUnique_  = fTuple->isUnique_;
      fldDef->isPrimary_ = fTuple->isPrimary_;
      fldDef->charset    = fTuple->charset  ;
      list.append(fldDef);
    }

    while (NULL != (data = fIter.nextElement()))
    {
        if (((CFIELD*)data)->tblPtr_ == tptr)
        {
            //add the information to the field list
             CFIELD
            CFIELD *fTuple = (CFIELD*)data;
            //FieldDef fldDef;
            m_field_def fldDef;

            ::memcpy(fldDef.fldName_,fTuple->fldName_,fTuple->fldNameRealLen) ;
            //fldDef.fldName_[IDENTIFIER_LENGTH] = '\0';
            fldDef.namelen    = fTuple->fldNameRealLen ;
            fldDef.type_      = fTuple->type_;
            fldDef.length_    = fTuple->length_;
            fldDef.isDefault_ = fTuple->isDefault_;

            os::memcpy(fldDef.defaultValueBuf_, fTuple->defaultValueBuf_,
                                         DEFAULT_VALUE_BUF_LENGTH);

            fldDef.isNull_    = fTuple->isNull_;
            fldDef.isUnique_  = fTuple->isUnique_;
            fldDef.isPrimary_ = fTuple->isPrimary_;
            list.append(fldDef);
        }
    }
    return;
}
*/
BackEnd::DbRetVal CatalogTableFIELD::getFieldPtrs(const char* fName,
                                                  void *tptr,
                                                  char **&fptr)
{
    if ( !fName)return BackEnd::ER_BadArg ;

    Chunk *fChunk = systemDatabase_->getSystemDatabaseChunk(FieldTableId);
    int i=0;

    bool found = false;

    void *data  = NULL;
    DbRetVal rv =BackEnd::OK;

    ChunkIterator fIter = fChunk->getIterator();
    found = false;
    while (NULL != (data = fIter.nextElement()))
    {
          if (((CFIELD*)data)->tblPtr_ == tptr)
        {
          if(!my_strcasecmp(system_charset_info,(char*)((CFIELD*)data)->fldName_,
                                         fName))
          {
             found = true;
             fptr[i++] = (char*) data;
             break;
          }
        }
    }
     if (!found)
    {
      return BackEnd::ER_NotFound;
    }
    return rv;
}
/*
 * SYNOPSIS  insert
 *
 *
 *
 *
 */
BackEnd::DbRetVal CatalogTableINDEX::insert(const char *name,
                                             void * tptr,
                                             int numFlds,
                                             bool  isUnique,
                                             void* chunk,
                                             int   bucketSize,
                                             void *hChunk,
                                             void *& tupleptr,
                                             IndexType _idxtype)
{
    Chunk *tChunk = systemDatabase_->getSystemDatabaseChunk(IndexTableId);
    ChunkIterator iter = tChunk->getIterator();
    /*Checking for index having same name, proceed further only
      if no such index exists*/
    char _address[BUFFERSIZE];
    sprintf(_address,"%p",tptr);
    Tbl_Field* tblfieldptr = systemDatabase_->searchTblFieldPtr
                            ((unsigned char*)_address,(uint32_t)strlen(_address));

    if (!tblfieldptr)
    {
      return BackEnd::ER_NotFound;
    }
    void *data = 0;
    BackEnd::ListIterator idx_iter = tblfieldptr->idxList.getIterator();

    while(idx_iter.hasElement())
    {
      data = (CINDEX*) idx_iter.nextElement();
      if (!my_strcasecmp(system_charset_info,((CINDEX*)data)->indName_, name))
      {
        return BackEnd::ER_Already;
      }
    }
    /*
    while ((data = iter.nextElement())!= NULL)
    {
         if (0 == strcmp(((CINDEX*)data)->indName_, name))
        {
           printError(BackEnd::ER_Already, "Index with name \'%s\' already exists "
           "on the table \'%s\'.", name, ((TABLE *)tptr)->tblName_);
           return BackEnd::ER_Already;
        }

    }*/

    BackEnd::DbRetVal rv =BackEnd::OK;
    // allocate.
    tupleptr = tChunk->allocate(systemDatabase_, &rv);
    if (!tupleptr)
    {
      return rv;
    }
    CINDEX *indexInfo          = (CINDEX*)tupleptr;
    indexInfo->init();
    strcpy(indexInfo->indName_, name);
    indexInfo->tblID_         = -1; //Not used currently
    /*the CTABLE pointer */
    indexInfo->tblPtr_        = tptr;

    indexInfo->numFlds_       = numFlds;

    indexInfo->indexType_     = _idxtype;
    /* chunk ptr*/
    indexInfo->chunkPtr_      = chunk;
    /* hash node ptr */
    indexInfo->hashNodeChunk_ = hChunk;

    indexInfo->noOfBuckets_   = bucketSize;

    indexInfo->isUnique_      = isUnique;
    /*Todo: charset*/

    /* append it at the tail of idxList*/
    tblfieldptr->idxList.append(tupleptr);
    errmsg_printf(ERRMSG_LVL_DBUG,
               _("[%s] line: [%d] One Row inserted into CINDEX : address [%p],index name:[%s]\n"),
                 __FILE__,__LINE__,tupleptr,name);
    return BackEnd::OK;
}

/**
 *  SYNOPSIS  insert
 *
 *  @param  name     char*
 *  @param  idxinfo  IndexInfo*   the pointer to IndexInfo
 *  @param  chunk
 *  @param  hChunk   Hash node chunk ptr
 *  @param  cinxptr  the out parameter.
 *
 */



BackEnd::DbRetVal CatalogTableINDEX::insert(const char *name,
                                             void * tptr,
                                             IndexInfo* idxinfo,
                                             void* chunk,
                                             void *hChunk, /*hash node chunk*/
                                             void *& cinxptr)
{
    Chunk *tChunk = systemDatabase_->getSystemDatabaseChunk(IndexTableId);
    ChunkIterator iter = tChunk->getIterator();
    char _address[BUFFERSIZE];
    sprintf(_address,"%p",tptr);
    Tbl_Field* tblfieldptr = systemDatabase_->searchTblFieldPtr
                             ((unsigned char*)_address,(uint32_t)strlen(_address));
    if (!tblfieldptr)
    {
      return BackEnd::ER_NotFound;
    }
    void *data = 0;
    BackEnd::ListIterator idx_iter = tblfieldptr->idxList.getIterator();

      while(idx_iter.hasElement())
    {
      data = (CINDEX*) idx_iter.nextElement();
      if (!my_strcasecmp(system_charset_info,((CINDEX*)data)->indName_, name))
      {
        return BackEnd::ER_Already;
      }
    }

    BackEnd::DbRetVal rv =BackEnd::OK;
    // allocate.
    cinxptr = tChunk->allocate(systemDatabase_, &rv);
    if (!cinxptr)
    {
        return rv;
    }
    CINDEX *indexInfo         = (CINDEX*)cinxptr;
    indexInfo->init();
    strcpy(indexInfo->indName_, name);
    indexInfo->tblID_         = -1; //Not used currently
    /*the table ptr */
    indexInfo->tblPtr_        = tptr;

    indexInfo->numFlds_       = 1+ idxinfo->sublist.size() ;

    indexInfo->indexType_     = idxinfo->indType;
    /* chunk ptr*/
    indexInfo->chunkPtr_      = chunk;

    indexInfo->hashNodeChunk_ = hChunk;
    if(idxinfo->indType == hashIndex) {
      indexInfo->noOfBuckets_   = ((HashIndexInfo*)idxinfo)->noOfBuckets;
    }
    indexInfo->isUnique_      = idxinfo->isUnique  ;

    /* charset */
    indexInfo->charset        = idxinfo->charset   ;

    indexInfo->flag           = idxinfo->flag;

    indexInfo->start          = idxinfo->start;

    indexInfo->fldOffset      = idxinfo->fldOffset;
    indexInfo->null_pos       = idxinfo->null_pos;
    indexInfo->bit_pos        = idxinfo->bit_pos;
    indexInfo->length         = idxinfo->length;
    errmsg_printf(ERRMSG_LVL_DBUG,
                   _("[%s] line: [%d] One Row inserted into CINDEX: Address [%p],Index name:[%s]\n"),
                     __FILE__,__LINE__,cinxptr,name);
    /* append it at the tail of idxList*/
    tblfieldptr->idxList.append(cinxptr);

    return BackEnd::OK;
}



DbRetVal CatalogTableINDEX::remove(const char *name,
                                   void * tptr ,
                                   void *&out_chunk,
                                   void *&out_hchunk,
                                   void *&out_ptr)
{
    Chunk *fChunk = systemDatabase_->getSystemDatabaseChunk(IndexTableId);
    ChunkIterator iter = fChunk->getIterator();
    void *data         = 0;
    Tbl_Field* tblfieldptr = 0;
    char _address[BUFFERSIZE];
    sprintf(_address,"%p",tptr);
    tblfieldptr   = systemDatabase_->searchTblFieldPtr
                            ((unsigned char*)_address,(uint32_t)strlen(_address));

    if (!tblfieldptr &&  tptr )
    {
      return BackEnd::ER_SysInternal;
    }
    out_ptr = 0;
    BackEnd::ListIterator idx_iter = tblfieldptr->idxList.getIterator();
      while(idx_iter.hasElement())
    {
       data = (CINDEX*) idx_iter.nextElement();
       if (!my_strcasecmp(system_charset_info,((CINDEX*)data)->indName_, name))
       {
          out_chunk  = (Chunk*)  ((CINDEX*)data)->chunkPtr_;
          out_hchunk = (Chunk*)  ((CINDEX*)data)->hashNodeChunk_;
          out_ptr= (void*) data;
          break ;
       }
    }
    /*
    while ((data = iter.nextElement())!= NULL)
    {
         if (0 == strcmp(((CINDEX*)data)->indName_, name))
         {
             //remove this element and store the tuple ptr
             //there will be only one row for this table(Primary key)
             chunk = (Chunk*) ((CINDEX*)data)->chunkPtr_;
             hchunk = (Chunk*) ((CINDEX*)data)->hashNodeChunk_;
             iptr = (void*) data;
             break;
         }
    }*/
    if (out_ptr)
    {   /*remove from the Hash Table*/
        if(tblfieldptr)
           tblfieldptr->idxList.remove(out_ptr),
        fChunk->free(systemDatabase_, out_ptr);

    }else
    {
        return ER_NotExists;
    }
    return OK;
}
DbRetVal CatalogTableINDEX::get(const char *name,
                                void *&chunk,
                                void *&hchunk,
                                void *&iptr)
{
    Chunk *fChunk = systemDatabase_->getSystemDatabaseChunk(IndexTableId);
    ChunkIterator iter = fChunk->getIterator();
    iptr  = 0 ;
    void *data = NULL;
    while ((data = iter.nextElement())!= NULL)
    {
         if (0 == strcmp(((CINDEX*)data)->indName_, name))
         {
             //remove this element and store the tuple ptr
             //there will be only one row for this table(Primary key)
             chunk  = (Chunk*) ((CINDEX*)data)->chunkPtr_;
             hchunk = (Chunk*) ((CINDEX*)data)->hashNodeChunk_;
             iptr = (void*) data;
             break;
         }
    }
    if (!iptr)
    {
      return ER_NotExists;
    }
    return OK;
}

uint32_t CatalogTableINDEX::getNumIndexes(void *tptr)
{
    Chunk *fChunk = systemDatabase_->getSystemDatabaseChunk(IndexTableId);

    ChunkIterator iter = fChunk->getIterator();

    uint32_t numIndex =0;
    // We search the Hash table by the tptr.
    Tbl_Field* tblfieldptr = 0;
    char _address[BUFFERSIZE];
    sprintf(_address,"%p",tptr);
    tblfieldptr   =    systemDatabase_->searchTblFieldPtr
                       ((unsigned char*)_address,(uint32_t)strlen(_address));
    if (tblfieldptr) {
      numIndex =tblfieldptr->idxList.size();
    }
    /*
    //void *iptr = NULL;
    while (NULL != (iptr = iter.nextElement()))
    {
         if (((CINDEX*)iptr)->tblPtr_ == tptr) numIndex++;
    }*/
    return numIndex;
}

char* CatalogTableINDEX::getIndexName(void *tptr, int32_t position)
{
    if (position <=0) return NULL;
    Chunk *fChunk = systemDatabase_->getSystemDatabaseChunk(IndexTableId);
    ChunkIterator iter = fChunk->getIterator();
    void *iptr   = NULL;
    char _address[BUFFERSIZE];
    sprintf(_address,"%p",tptr);
    Tbl_Field* tblfieldptr = 0;
    tblfieldptr   =    systemDatabase_->searchTblFieldPtr
                             ((unsigned char*)_address,(uint32_t)strlen(_address));
    if (tblfieldptr) {
      iptr = tblfieldptr->idxList.get(position);
      if (iptr) return ((CINDEX*)iptr)->indName_;
    }
    /*
    //int numIndex =0;
    //int curPos   =0;
    while (NULL != (iptr = iter.nextElement()))
    {
         if (((CINDEX*)iptr)->tblPtr_ == tptr) curPos++;
         if ( curPos == position ) return ((CINDEX*)iptr)->indName_;
    }
    */
    return NULL;

}

void CatalogTableINDEX::getIndexPtrs(void *tptr, char **&array)
{
    int i=0;
    void *iptr = NULL;
    char _address[BUFFERSIZE];
    sprintf(_address,"%p",tptr);
    Tbl_Field* tblfieldptr = 0;
    tblfieldptr   =  systemDatabase_->searchTblFieldPtr
                     ((unsigned char*)_address,(uint32_t)strlen(_address));

    if (tblfieldptr) {
      BackEnd::ListIterator idx_iter = tblfieldptr->idxList.getIterator();
      while(idx_iter.hasElement())
     {
       iptr = (CINDEX*) idx_iter.nextElement();
       array[i++] = (char*) iptr;
     }
    }
    /*


    Chunk *fChunk = systemDatabase_->getSystemDatabaseChunk(IndexTableId);

    ChunkIterator iter = fChunk->getIterator();
    while (NULL != (iptr = iter.nextElement()))
    {
         if (((CINDEX*)iptr)->tblPtr_ == tptr)
       {
         array[i++] = (char*) iptr;
       }
    }
    */
    return;
}

CINDEX* CatalogTableINDEX::getIndexPtrByName(void *tptr, const char* idxname)
{
  CINDEX*iptr             =    NULL;
  Tbl_Field* tblfieldptr  =    0;
  char _address[BUFFERSIZE];
  sprintf(_address,"%p",tptr);
  tblfieldptr =systemDatabase_->searchTblFieldPtr
                               ((unsigned char*)_address,(uint32_t)strlen(_address));

  if (tblfieldptr) {
      BackEnd::ListIterator idx_iter = tblfieldptr->idxList.getIterator();
      while(idx_iter.hasElement())
    {
       iptr = (CINDEX*) idx_iter.nextElement();
       if (!my_strcasecmp(system_charset_info,((CINDEX*)iptr)->indName_,idxname))
      { // found it.
        return iptr;
      }
    }
  }
  return NULL ;
}

ChunkIterator CatalogTableINDEX::getIterator(void *iptr)
{
    CINDEX *index = (CINDEX*)iptr;
    return ((Chunk*)index->chunkPtr_)->getIterator();
}


int CatalogTableINDEX::getNoOfBuckets(void *iptr)
{
    CINDEX *index = (CINDEX*)iptr;
    return index->noOfBuckets_;
}

int CatalogTableINDEX::getUnique(void *iptr)
{
    CINDEX *index = (CINDEX*)iptr;
    return index->isUnique_;
}
char* CatalogTableINDEX::getName(void *iptr)
{
    CINDEX *index = (CINDEX*)iptr;
    return index->indName_;
}

DbRetVal CatalogTableUSER::insert(const char *, const char *)
{  /*
    Chunk *tChunk = systemDatabase_->getSystemDatabaseChunk(UserTableId);
    DbRetVal rv = OK;
    USER *usrInfo = (USER*)tChunk->allocate(systemDatabase_, &rv);
    if (NULL == usrInfo)
    {
        printError(rv,
                   "Could not allocate for USER catalog table");
        return rv;
    }
    strcpy(usrInfo->userName_, name);
    // TODO strcpy(usrInfo->password_, OS::encrypt(pass, "A0"));
     *
     */
    return OK;

}
/*
DbRetVal CatalogTableUSER::authenticate(const char *name, const char *pass,
                                   bool &isAuthenticated, bool &isDba)
{
    Chunk *tChunk = systemDatabase_->getSystemDatabaseChunk(UserTableId);
    ChunkIterator iter = tChunk->getIterator();
    void *data = NULL;
    while (NULL != (data = iter.nextElement()))
    {
        if (strcmp(((USER*)data)->userName_, name) == 0)
        {
            //verify the password
            char * enpass = os::encrypt(pass,"A0");
            if (0 == strcmp(enpass, ((USER*)data)->password_))
            {
                isAuthenticated = true;
                if (0 == strcmp(((USER*)data)->userName_, DBAUSER))
                    isDba = true; else isDba = false;
                return OK;
            }
        }
    }
    isAuthenticated = false;
    return OK;
}
*/
DbRetVal CatalogTableUSER::remove(const char *)
{   /*
    Chunk *tChunk = systemDatabase_->getSystemDatabaseChunk(UserTableId);
    ChunkIterator iter = tChunk->getIterator();
    void *data = NULL;
    while ((data = iter.nextElement())!= NULL)
    {
        if (strcmp(((CUSER*)data)->userName_, name) == 0)
        {
            //remove this element
            tChunk->free(systemDatabase_, data);
            return OK;
        }
    }
    printError(ER_NotExists,"User %s not exists in catalog table", name);
    */
    return ER_NotExists;
}

/*
DbRetVal CatalogTableUSER::changePass(const char *name, const char *pass)
{
    Chunk *tChunk = systemDatabase_->getSystemDatabaseChunk(UserTableId);
    ChunkIterator iter = tChunk->getIterator();
    void *data = NULL;
    while (NULL != (data = iter.nextElement()))
    {
        if (strcmp(((USER*)data)->userName_, name) == 0)
        {
            //change the password
            strcpy(((USER*)data)->password_, os::encrypt(pass, "A0"));
            return OK;
        }
    }
    printError(ErrNotExists,"User %s not exists in catalog table", name);
    return ErrNotExists;
}
*/
